package com.mjf.orderpay_detect

import java.net.URL

import com.mjf.dim.{OrderEvent, ReceiptEvent}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 实时对账
 */
object OrderPayTxMatch {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val orderResource: URL = getClass.getResource("/OrderLog.csv")
    val orderStream: DataStream[String] = env.readTextFile(orderResource.getPath)

    // 订单数据
    val orderEventStream: DataStream[OrderEvent] = orderStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          OrderEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })
      .filter(_.eventType != "") // 过滤出pay事件
      .keyBy(_.txId)

    val receiptResource: URL = getClass.getResource("/ReceiptLog.csv")
    val receiptStream: DataStream[String] = env.readTextFile(receiptResource.getPath)

    // 实时到账数据
    val ReceiptEventStream: DataStream[ReceiptEvent] = receiptStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          ReceiptEvent(dataArr(0), dataArr(1), dataArr(2).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ReceiptEvent](Time.seconds(3)) {
        override def extractTimestamp(element: ReceiptEvent): Long = element.timestamp * 1000L
      })
      .keyBy(_.txId)

    // 用 connect 连接两条流，匹配事件进行处理
    val resultStream : DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream
      .connect(ReceiptEventStream)
        .process(new OrderPayTxDetect())

    val unmatchedPays: OutputTag[OrderEvent] = new OutputTag[OrderEvent]("unmatched-pays")
    val unmatchedReceipts: OutputTag[ReceiptEvent] = new OutputTag[ReceiptEvent]("unmatched-receipts")

    resultStream.print("matched")
    resultStream.getSideOutput(unmatchedPays).print("unmatched-pays")
    resultStream.getSideOutput(unmatchedReceipts).print("unmatched-receipts")

    env.execute("OrderPayTxMatch")

  }
}

// 实现两条流数据的匹配检验
class OrderPayTxDetect() extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {

  // 用两个ValueState,保存当前交易对应的支付事件和到账事件
  lazy val payState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("pay", classOf[OrderEvent]))
  lazy val receiptState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))

  val unmatchedPays: OutputTag[OrderEvent] = new OutputTag[OrderEvent]("unmatched-pays")
  val unmatchedReceipts: OutputTag[ReceiptEvent] = new OutputTag[ReceiptEvent]("unmatched-receipts")

  override def processElement1(pay: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // pay来了，考察有没有对应的receipt来过
    val receipt: ReceiptEvent = receiptState.value()
    if(receipt != null) {
      // 如果已经有receipt，那么正常匹配，输出到主流
      out.collect((pay, receipt))
      receiptState.clear()
    } else {
      // 如果receipt还没来，那么把pay存入状态，注册定时器等待5秒
      payState.update(pay)
      ctx.timerService().registerEventTimeTimer(pay.eventTime * 1000L + 5000L)
    }
  }

  override def processElement2(receipt: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // receipt来了，考察有没有对应的pay来过
    val pay: OrderEvent = payState.value()
    if(pay != null) {
      // 如果已经有pay，那么正常匹配，输出到主流
      out.collect((pay, receipt))
      payState.clear()
    } else {
      // 如果pay还没来，那么把receipt存入状态，注册定时器等待5秒
      receiptState.update(receipt)
      ctx.timerService().registerEventTimeTimer(receipt.timestamp * 1000L + 3000L)
    }
  }

  // 触发定时器有两种情况，所以要判断当前有没有pay和receipt
  override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 如果pay不为空，说明receipt没来, 输出unmatchedPays
    if(payState.value() != null) {
      ctx.output(unmatchedPays, payState.value())
    }

    if(receiptState.value() != null) {
      ctx.output(unmatchedReceipts, receiptState.value())
    }

    // 清空状态
    payState.clear()
    receiptState.clear()
  }

}
